package com.navinfo.platform.thousandkilo.utils;

import com.navinfo.platform.hbase.api.HbaseTemplate;
import com.navinfo.platform.thousandkilo.dto.ThousandKiloData;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Mutation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

public class BigDataUtils implements Serializable {
    private static final Logger logger = LoggerFactory.getLogger(BigDataUtils.class);
    private String port;
    private String quorum;
    private String parent;
    private String tableName;

    public String getTableName() {
        return tableName;
    }
    public void setTableName(String tableName) {
        this.tableName = tableName;
    }
    public String getQuorum() {
        return quorum;
    }
    public void setQuorum(String quorum) {
        this.quorum = quorum;
    }
    public String getPort() {
        return port;
    }
    public void setPort(String port) {
        this.port = port;
    }
    public String getParent() {
        return parent;
    }
    public void setParent(String parent) {
        this.parent = parent;
    }

    public void saveToHbase(Collection<ThousandKiloData> collection) {
        long start = System.currentTimeMillis();
        logger.info("[saveToHbase]start");

        List<Mutation> saveOrUpdates = new ArrayList<>();
        for (ThousandKiloData data : collection) {
            if (data != null) {
                saveOrUpdates.add(HbaseConverter.toPut(data));
            }
        }
        logger.info("[saveToHbase]saveOrUpdates.size:{}", saveOrUpdates.size());
        if (saveOrUpdates.size() > 0) {
            createHbaseTemplate().saveOrUpdates(getTableName(), saveOrUpdates);
        }
        logger.info("[saveToHbase]end cost:{}ms", System.currentTimeMillis() - start);
    }

    public ThousandKiloData queryHbase(String row) {
        try {
            return createHbaseTemplate().get(getTableName(), row, new HbaseConverter());
        } catch (Exception e) {
            logger.info(e.getMessage(), e);
            return null;
        }
    }

    private HbaseTemplate createHbaseTemplate() {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", getQuorum());
        configuration.set("hbase.zookeeper.property.clientPort", getPort());
        configuration.set("zookeeper.znode.parent", getParent());
        return new HbaseTemplate(configuration);
    }
}
